package com.hadoop.seq;


import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.lib.InputSampler;
import org.apache.hadoop.mapred.lib.TotalOrderPartitioner;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/*
 * 전체 정렬은 하나의 파티션만 사용해서 해결 -> 입력 데이터의 키를 기준으로 정렬하므로 단일 파티션에 모든 데이터가 정렬되어 출력됨
 * (리듀스 태스크가 실행되는 데이터 노드만 부하가집중 --> 분산 처리)
 * 1. 입력 데이터를 샘플링해서 데이터의 분포도 조사
 * 2. 데이터의 분포도에 맞게 파티션 정보를 미리 생성
 * 3. 미리 생성한 파티션 정보에 맞게 출력 데이터를 생성
 * 4. 각 출력 데이터 병합
 * 
 */
public class SequenceFileTotalSort extends Configured implements Tool {

	public int run(String[] args) throws Exception {
		JobConf conf = new JobConf(getConf(), SequenceFileTotalSort.class);
		conf.setJobName("SequenceFileTotalSort");

		//항공운항 기준으로 생성된 sequence 파일
		conf.setInputFormat(SequenceFileInputFormat.class);
		conf.setOutputFormat(SequenceFileOutputFormat.class);
		conf.setOutputKeyClass(IntWritable.class); //항공 운항 기준으로 정렬돼야 하기 때문에 
		conf.setPartitionerClass(TotalOrderPartitioner.class); //파티션 정보를 직접 설정해야 하기 때문에 TotalOrderPartitoner를 설정.

		// SequenceFile 압축 포맷 설정
		SequenceFileOutputFormat.setCompressOutput(conf, true);
		SequenceFileOutputFormat
				.setOutputCompressorClass(conf, GzipCodec.class);
		SequenceFileOutputFormat.setOutputCompressionType(conf,
				CompressionType.BLOCK);

		// 입출력 설정 
		FileInputFormat.setInputPaths(conf, new Path(args[0]));
		FileOutputFormat.setOutputPath(conf, new Path(args[1]));

		// 파티션을 어떻게 구성할지 정의해줌. 
		Path inputDir = FileInputFormat.getInputPaths(conf)[0];
		inputDir = inputDir.makeQualified(inputDir.getFileSystem(conf));
		Path partitionFile = new Path(inputDir, "_partitions"); //파티션 경로는 "입력데이터경로_partitions"라고 설정
		
		//파티션 경로 서정 (TotalOrderPartitioner는 이 경로에 키 정보를 저장)
		TotalOrderPartitioner.setPartitionFile(conf, partitionFile);

		//입력 데이터 샘플링
		InputSampler.Sampler<IntWritable, Text> sampler = new InputSampler.RandomSampler<IntWritable, Text>(
				0.1, 1000, 10); //10개의 입력 스플릿에서 0.1%의 확률로 1,000건의 데이터를 샘플링
		InputSampler.writePartitionFile(conf, sampler);

		//task가 파티션 정보를 참조할 수 있게 분산 캐시에 파티션 정보를 등록.
		URI partitionUri = new URI(partitionFile.toString() + "#_partitions");
		DistributedCache.addCacheFile(partitionUri, conf);
		DistributedCache.createSymlink(conf);

		JobClient.runJob(conf);

		return 0;
	}

	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(),
				new SequenceFileTotalSort(), args);
		System.out.println("## RESULT:" + res);
	}
}
